package com.lagou.interceptor;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.commons.compress.utils.Charsets;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.interceptor.Interceptor;

import java.time.Instant;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.time.format.DateTimeFormatter;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;

public class CustomerInterceptor implements Interceptor {
    @Override
    public void initialize() {

    }

    /**
     * 逐条处理Event
     * @param event 事件
     * @return 事件
     */
    @Override
    public Event intercept(Event event) {
        //获取 event 的 body
        String eventBody = new String(event.getBody(), Charsets.UTF_8);
        //获取 event 的 header
        Map<String, String> headersMap = event.getHeaders();
        //解析body获取json串
        String[] bodyArr = eventBody.split("\\s+");

        try {
            String jsonStr = bodyArr[6];
            //解析json串获取时间戳
            JSONObject jsonObject = JSON.parseObject(jsonStr);
            String timestampStr = jsonObject.getJSONObject("app_active").getString("time");

            //将时间戳转换为字符串
            //将字符串转换为Long
            long timestamp = Long.parseLong(timestampStr);
            DateTimeFormatter timeFormatter = DateTimeFormatter.ofPattern("yyyy-MM-dd");
            Instant instant = Instant.ofEpochMilli(timestamp);
            LocalDateTime localDateTime = LocalDateTime.ofInstant(instant, ZoneId.systemDefault());
            String date = timeFormatter.format(localDateTime);
            //将转换后的字符串放置header中
            headersMap.put("logtime", date);
            event.setHeaders(headersMap);
        } catch (NumberFormatException e) {
            headersMap.put("logtime", "Unknown");
            event.setHeaders(headersMap);
            e.printStackTrace();
        }
        return event;
    }

    @Override
    public List<Event> intercept(List<Event> events) {
        List<Event> lstEvent = new ArrayList<>();
        for (Event event : events) {
            Event intercept = intercept(event);
            if (intercept != null) {
                lstEvent.add(intercept);
            }
        }
        return lstEvent;
    }

    @Override
    public void close() {

    }

    public static class Builder implements Interceptor.Builder {

        @Override
        public Interceptor build() {
            return new CustomerInterceptor();
        }

        @Override
        public void configure(Context context) {

        }
    }
}
